package report

import java.lang

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.functions.MapFunction
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat
import org.apache.flink.api.scala.{DataSet, ExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.types.Row

object AppAnalysis {
  def main(args: Array[String]): Unit = {
    //创建运行环境上下文
    val env = ExecutionEnvironment.getExecutionEnvironment
    //读取数据
    val sourceDataSet: DataSet[String] = env.readTextFile("C:\\Users\\thinkpad\\Desktop\\data.txt")
    //数据清洗，filter
    val filted: DataSet[String] = sourceDataSet.filter(line => line.substring(line.indexOf("["), line.indexOf("]")).length > 1)
    //统计数据，拿出appname，上行流量，下行流量
    val dataResult: DataSet[(String, Double, Double)] = filted.flatMap(line => {
      //将line转换成json对象
      val jsonObject = JSON.parseObject(line)
      //获取traffic
      val traffic = jsonObject.getJSONObject("traffic")
      //获取trafficInfo
      val trafficInfoArr = traffic.getJSONArray("trafficInfo")
      var list = List[(String, Double, Double)]()
      //遍历循环数组
      for (i <- 0 until trafficInfoArr.size()) {
        //获取数组中的每一个元素
        val trafficInfo: JSONObject = trafficInfoArr.getJSONObject(i)
        //获取appname
        val ispname = trafficInfo.getString("ispname")
        //获取上行流量
        val ispuptraffic: lang.Double = trafficInfo.getDouble("ispuptraffic")
        //获取下行流量
        val ispdowntraffic = trafficInfo.getDouble("ispdowntraffic")
        //zhujie
        list :+= (ispname, ispuptraffic.toDouble, ispdowntraffic.toDouble)
      }
      list
    })
    //.groupBy(0).sum(1).print()
    //聚合 reducebykey 单独聚合
    val upSum: AggregateDataSet[(String, Double)] = dataResult.map(tp => (tp._1, tp._2)).groupBy(0).sum(1)
    val downSum: AggregateDataSet[(String, Double)] = dataResult.map(tp => (tp._1, tp._3)).groupBy(0).sum(1)
    val joined: JoinDataSet[(String, Double), (String, Double)] = upSum.join(downSum).where(tp => tp._1).equalTo(tp => tp._1)
    val result: DataSet[Row] = joined.map(new MapFunction[((String, Double), (String, Double)), Row] {
      override def map(t: ((String, Double), (String, Double))): Row = {
        val row = new Row(4)
        row.setField(0, t._1._1.getBytes("utf-8"))
        row.setField(1, t._1._2)
        row.setField(2, t._2._2)
        row.setField(3, t._1._2 + t._2._2)
        row
      }
    })


    val driverClass = "com.mysql.jdbc.Driver"
    val dbUrl = "jdbc:mysql://localhost:3306/test"
    val userNmae = "root"
    val passWord = "root"
    val sql = "insert into test.appLiuLiang (appname,up,down,sum) values (?,?,?,?)"
    //写入mysql write.jdbc
    result.output(JDBCOutputFormat.buildJDBCOutputFormat()
      .setDrivername(driverClass)
      .setDBUrl(dbUrl)
      .setUsername(userNmae)
      .setPassword(passWord)
      .setQuery(sql)
      .finish()
    )

    //启动执行
    env.execute()
  }
}
